<?php
include_once (dirname(__FILE__) . '/../init.php');

class QueueConsumerManagerJob extends CronJob {
	private $workers = array();

  //保证至少有X个consumer在并行跑
	private static $parallel_limit = array(
	);

	protected function startWorkers() {
		$files = glob(__DIR__ . '/consumer/*.php');
		foreach ($files as $_file) {
			$name = basename($_file, '.php');
			$paral_num = isset(self::$parallel_limit[$name]) ? self::$parallel_limit[$name] : 1;
			for ($i = 0; $i < $paral_num; $i++) {
				$this->workers[$name . $i] = pcntl_fork();
				if ($this->workers[$name . $i] === 0) { //子进程进入了这个岔路口，父进程直接执行if后面的代码
					$cmd = "php {$_file} > /dev/null 2>>" . LOG_DIR . "/consumer.err";
					$this->runCmd($cmd);
					posix_kill(posix_getpid(), SIGTERM);
				}
				$this->log("{$name}({$i}) started! " . count($this->workers));
			}
		}
		while (count($this->workers)) {
			$pid = pcntl_wait($status); //父进程中可以拿到子进程的ID
			$_key = array_search($pid, $this->workers);
			if ($_key !== false) {
				unset($this->workers[$_key]);
				$this->log("{$_key} stopped！ " . count($this->workers));
			}
			usleep(50000);
		}
	}

	private function runCmd($cmd) {
		passthru($cmd, $return);
		if ($return == QueueConsumer::SIGNAL_RESTART) {
			$this->log("restart cmd: {$cmd}");
			$this->runCmd($cmd); //restart
			$this->log("!! Consumer restart !!: {$cmd}");
		}
	}
}

if (ON_PRODUCTION) {
	$q = new QueueConsumerManagerJob('startWorkers');
	$q->doJob();
}
